package com.world.rickcloudendpoint.demo.listener;

import com.world.rickcloudendpoint.demo.MsgConsumer;
import com.world.rickcloudendpoint.demo.local.LocalMqConsumerContainer;
import com.world.rickcloudendpoint.demo.local.LocalQueueFactory;
import lombok.extern.log4j.Log4j2;

import java.util.AbstractQueue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @description:
 * @author: jlsong
 * @date: 2022/06/03 20:25:23
 * @version: 1.0
 */
@Log4j2
public class LocalQueueListener implements Runnable {
    private LocalQueueFactory queueFactory;
    private String queueName;
    private MsgConsumer consumer;

    public LocalQueueListener(LocalQueueFactory queueFactory, String queueName, MsgConsumer consumer) {
        this.queueFactory = queueFactory;
        this.queueName = queueName;
        this.consumer = consumer;
    }

    @Override
    public void run() {
        log.info("Local queue listener 正在启动...queue: {} ", queueName);
        while (LocalMqConsumerContainer.run) {
            try {
                // brpop 阻塞到有消息来才拉取消息
                AbstractQueue queue = queueFactory.getQueue(queueName);
                LinkedBlockingQueue blockingQueue = (LinkedBlockingQueue) queue;
                List<Object> collects = new ArrayList<>();
                take(blockingQueue, collects, 2000, 5, TimeUnit.MILLISECONDS);
                // process list.
                if (collects == null || collects.size() == 0) {
                    continue;
                }
                // 处理消息
                try {
                    log.info("{}=>{}", queueName, collects.size());
                    consumer.onMessage(collects);
                } catch (Exception e) {
                    consumer.onError(collects, e);
                }

            } catch (Exception e) {
                if (LocalMqConsumerContainer.run) {
                    log.error("异常队列：{},原因：{}", queueName, e.getMessage(), e);
                } else {
                    log.info("Local queue listener 退出.....queue: {} ,原因：{}", queueName, e.getMessage(), e);
                }
            }
        }
    }

    static RelativeTimeProvider timeProvider = new SystemRelativeTimeProvider();

    public static <T> int take(BlockingQueue<T> queue, Collection<? super T> batch, int maxBatchSize, long maxLatency, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(maxLatency);
        int curBatchSize = 0;
        long stopBatchTimeNanos = -1;

        // The loop flow is 1) block, 2) drain queue, 3) possibly consume batch.
        while (true) {
            boolean timeout = false;
            if (stopBatchTimeNanos == -1) {
                // Start of new batch. Block for the first item of this batch.
                batch.add(queue.take());
                curBatchSize++;
                stopBatchTimeNanos = timeProvider.relativeTime(TimeUnit.NANOSECONDS) + nanos;
            } else {
                // Continue existing batch. Block until an item is in the queue or the
                // batch timeout expires.
                T element = queue.poll(stopBatchTimeNanos - timeProvider.relativeTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
                if (element == null) {
                    // Timeout occurred.
                    break;
                }
                batch.add(element);
                curBatchSize++;
            }
            curBatchSize += queue.drainTo(batch, maxBatchSize - curBatchSize);
            if (curBatchSize >= maxBatchSize) {
                // End current batch.
                break;
            }
        }
        return curBatchSize;
    }
}
